package com.poetic.network.flow;

import com.poetic.network.flow.domain.UserBehavior;
import com.poetic.network.flow.enums.BehaviorEnum;
import com.poetic.network.flow.function.UVCountByWindow;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

/**
 * <pre>
 *  TODO
 * Created by lianghuikun on 2020-09-15.
 * </pre>
 * 。UV
 * 指的是一段时间（比如一小时）内访问网站的总人数，1 天内同一访客的多次访问
 * 只记录为一个访客。
 *
 * @author lianghuikun
 */
public class UniqueVisitorTask {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设定 Time 类型为 EventTime
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        // 为了打印到控制台的结果不乱序，我们配置全局的并发为 1，这里改变并发对结果正确性没有影响
        env.setParallelism(1);
        String path = "/Users/lianghuikun/indigo/poetic/flink/user-behavior-anylysis/data/UserBehavior.csv";
        env.readTextFile(path)
                .map(new MapFunction<String, UserBehavior>() {
                    @Override
                    public UserBehavior map(String value) throws Exception {
                        String[] data = value.split("\\,");
                        return UserBehavior.builder()
                                .userId(Long.valueOf(data[0]))
                                .itemId(Long.valueOf(data[1]))
                                .categoryId(Integer.valueOf(data[2]))
                                .behavior(data[3])
                                .timestamp(Long.valueOf(data[4]))
                                .build();
                    }
                })
                .filter(new FilterFunction<UserBehavior>() {
                    @Override
                    public boolean filter(UserBehavior value) throws Exception {
                        return value.getBehavior().equals(BehaviorEnum.pv.name());
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(1)
                ).withTimestampAssigner(new SerializableTimestampAssigner<UserBehavior>() {
                    @Override
                    public long extractTimestamp(UserBehavior element, long recordTimestamp) {
                        return element.getTimestamp() * 1000;
                    }
                }))
                // 不做keyBy的，用timeWindowAll
                // 统计一小时的UV
                .timeWindowAll(Time.hours(1))
                // 不建议用apply,建议用aggregate。前者是所有的然后聚合，后者是增量聚合
                .apply(new UVCountByWindow())
                .print();

        env.execute("Unique Visitor Job");
    }
}
